#pragma once

#define BUFFER_DEFAULT_SIZE 1024*1024//默认缓冲区大小为1M
#include <vector>
#include <string>
#include <cassert>
#include <cstring>
#include<iostream>
#include<cstdint>
#include<ctime>
#include<fcntl.h>
#include<sys/types.h>
#include<sys/socket.h>
#include<netinet/in.h>
#include<arpa/inet.h>
#include<unistd.h>
#include<functional>
#include<sys/epoll.h>
#include<errno.h>
#include<unordered_map>
#include<mutex>
#include<thread>
#include<memory>
#include<sys/eventfd.h>
#include<sys/timerfd.h>
#include<any>
#include<condition_variable>
#include<signal.h>



#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL ERR
#define LOG(level,format,...) do{\
    if(level<LOG_LEVEL)break;\
    time_t t=time(NULL);\
    struct tm*ltm=localtime(&t);\
    char tmp[32]={0};\
    strftime(tmp,32,"%Y-%m-%d %H:%M:%S",ltm);\
    fprintf(stdout,"[%p:%s:%s:%d]" format "\n",(void*)pthread_self(),tmp,__FILE__,__LINE__,##__VA_ARGS__ );\
    }while(0)
#define INF_LOG(format,...) LOG(INF,format,##__VA_ARGS__)
#define DBG_LOG(format,...) LOG(DBG,format,##__VA_ARGS__)
#define ERR_LOG(format,...) LOG(ERR,format,##__VA_ARGS__)



class Buffer{
    private:
        std::vector<char> _buffer;//用vector进行内存空间管理，便于扩容
        uint64_t _read_idx;//读偏移--下次要开始读取的位置
        uint64_t _write_idx;//写偏移--下一次要写入的位置
    public:
        //构造函数
        Buffer():_read_idx(0),_write_idx(0),_buffer(BUFFER_DEFAULT_SIZE,0){}
        char* Begin(){return &*_buffer.begin();}
        //获取当前写入起始地址
        char * WritePosition(){return Begin()+_write_idx;}
        //获取当前读取起始地址
        char* ReadPosition(){return Begin()+_read_idx;}
        //获取后沿空闲空间大小
        uint64_t TailIdleSize(){return _buffer.size()-_write_idx;}
        //获取前沿空闲空间大小
        uint64_t HeadIdleSize(){return _read_idx;}
        //获取可读数据大小
        uint64_t ReadAbleSize(){return _write_idx-_read_idx;}
         //将读偏移后移
        void  MoveReadOffset(uint64_t len){
            if(len==0)return;
            assert(len<=ReadAbleSize());
            _read_idx+=len;
       }
        //将写偏移后移
        void  MoveWriteOffset(uint64_t len){
            assert(len<=TailIdleSize());
            _write_idx+=len;
        }
        //确保可写空间足够--不够就把数据前移，还不够就扩容
        void EnsureWriteSpace(uint64_t len){
            //如果末尾空间足够，则直接返回
            if(TailIdleSize()>=len)return;
            //如果末尾空间不够，判断前后空余空间之和够不够
            if(len<=HeadIdleSize()+TailIdleSize()){
                //如果够，则把数据前移
                uint64_t rsz=ReadAbleSize();//保存当前数据大小
                std::copy(ReadPosition(),ReadPosition()+rsz,Begin());//把数据前移
                _read_idx=0;
                _write_idx=rsz;
            }
            else{//如果不够，则不移动数据，直接在后面扩容
                _buffer.resize(_write_idx+len);
            }
        }
        //写入数据
        void Write(const void* data, uint64_t len){
            if(len==0)return;
            EnsureWriteSpace(len);
            const char*d=(const char*)data;
            std::copy(d,d+len,WritePosition());
        }
        void WriteAndPush(const void *data, uint64_t len){
        Write(data,len);
        MoveWriteOffset(len);
        }
        void WriteString(const std::string& str){Write(str.c_str(),str.size());}
        void WriteStringAndPush(const std::string& str){WriteString(str);MoveWriteOffset(str.size());}
        void WriteBuff(Buffer& buff){Write(buff.ReadPosition(),buff.ReadAbleSize());}
        void WriteBuffAndPush(Buffer&buff){WriteBuff(buff);MoveWriteOffset(buff.ReadAbleSize());}
        //读取数据
        void Read(void* buf, uint64_t len){
            assert(len<=ReadAbleSize());
            std::copy(ReadPosition(),ReadPosition()+len,(char*)buf);
        }
        std::string ReadAsString(uint64_t len){
            std::string str(len,'\0');
            Read(&str[0],len);
            return str;
        }
        void ReadAndPop(void* buf, uint64_t len){
            Read(buf,len);
            MoveReadOffset(len);
        }
        std::string ReadAsStringAndPop(uint64_t len){
            std::string str=ReadAsString(len);
            MoveReadOffset(len);
            return str;
        }
        //寻找换行字符
        char* FindCRLF(){
            void*p=memchr(ReadPosition(),'\n',ReadAbleSize());
            return (char*)p;
        }
        //获取一行数据
        std::string GetLine(){
            void*p=FindCRLF();
            if(p==nullptr)return "";
            //+1是因为要把换行符也读进来
            uint64_t len=((char*)p)-ReadPosition()+1;
            return ReadAsString(len);
        }
        std::string GetLineAndPop(){
            std::string str=GetLine();
            MoveReadOffset(str.size());
            return str;
        }
        //清空缓冲区A
        void Clear(){
            _read_idx=0;
            _write_idx=0;
        }
};



#define Max_Listen 32//默认最大监听数
class Socket{
    private:
        int _sockfd;
    public:
    Socket(): _sockfd(-1){};
    Socket(int sockfd): _sockfd(sockfd){};
    ~Socket(){;}
    //获取套接字描述符
    int Fd(){return _sockfd;}
    //创建套接字
    bool Create(){
        _sockfd=socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);//注意这里
        if(_sockfd<0){
            ERR_LOG("create socket failed");
            return false;
        }
        DBG_LOG("create socket success");
        return true;

    }
    //绑定地址
    bool Bind(const std::string&ip,uint16_t port){
        struct sockaddr_in addr;
        addr.sin_family=AF_INET;
        addr.sin_port=htons(port);
        addr.sin_addr.s_addr=inet_addr(ip.c_str());
        socklen_t len=sizeof (addr);
        int ret=bind(_sockfd,(struct sockaddr*)&addr,len);
        if(ret<0){
            ERR_LOG("bind socket failed");
            return false;
        }
        return true;
    }
    //开始监听
    bool Listen(int backlog=Max_Listen){
        int ret=listen(_sockfd,backlog);
        if(ret<0){
            ERR_LOG("listen socket failed");
            return false;
        }
        return true;
    }
    //向服务器发起链接
    bool Connect(uint16_t port,const std::string&ip){
        DBG_LOG("connect to %s:%d",ip.c_str(),port);
        struct sockaddr_in addr;
        addr.sin_family=AF_INET;
        addr.sin_port=htons(port);
        addr.sin_addr.s_addr=inet_addr(ip.c_str());
        socklen_t len=sizeof (addr);
        int ret=connect(_sockfd,(struct sockaddr*)&addr,len);
        if(ret<0){
            ERR_LOG("connect socket failed");
            return false;
        }
        DBG_LOG("connect socket success");
        return true;
    }
    //获取链接
    int Accept(){
        int ret=accept(_sockfd,NULL,NULL);
        if(ret<0){
            ERR_LOG("accept socket failed");
            return -1;
        }
        return ret;
    }
    //接收数据
    ssize_t Recv(void*buf,size_t len,int flag=0){
       int ret= recv(_sockfd,buf,len,flag);
       if(ret<=0){
           if(errno==EAGAIN||errno==EWOULDBLOCK||errno==EINTR){//读完了或者被信号打断了
               return 0;
           }
           ERR_LOG("socket recv  failed fd is %d", _sockfd);
           sleep(1);
           return -1;
       }
       return ret;//接受到的信号长度
    }
    ssize_t NonBlockingRecv(void*buf,size_t len){
        if(len==0)return 0;
        return Recv(buf,len,MSG_DONTWAIT);
    }
    //发送数据
    ssize_t Send(const void*buf,size_t len,int flag=0){
        int ret= send(_sockfd,buf,len,flag);
        if(ret<0){
            if(errno==EAGAIN||errno==EWOULDBLOCK||errno==EINTR){//写完了或者被信号打断了
                return 0;
            }
            ERR_LOG("send socket failed %d:%s",errno,strerror(errno));
            return -1;
        }
        return ret;//发送的信号长度
    }
    ssize_t NonBlockingSend(const void*buf,size_t len){
        return Send(buf,len,MSG_DONTWAIT);
    }
    //关闭套接字
    void Close(){
        if(_sockfd!=-1){
            std::cout << "已关闭" << _sockfd << std::endl;
            close(_sockfd);
            _sockfd=-1;
        }
    }
    //创建一个客户端链接
    bool CreatClient(uint16_t port,const std::string&ip){
       //创建套接字
       if(!Create())return false;
       //连接服务器，不用手动bind，connect会自动bind
       if(!Connect(port,ip))return false;
       return true;
    }
    //创建一个服务端链接,默认阻塞，0阻塞，1非阻塞
    bool CreateServer(uint16_t port,const std::string&ip="0.0.0.0",bool none_block=false){
        //创建套接字
        if(!Create())return false;
         //启动地址重用
         ReuseAddress();
        //绑定地址
        if(!Bind(ip,port))return false;
        //开始监听
        if(!Listen())return false;
        //设置非阻塞
        if(none_block)
        NonBlocking();
        return true;
   }
    //开启端口地址重用
       void ReuseAddress(){
        int optval = 1;
        int ret1=setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval));
        int ret2=setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(optval));
        if(ret1<0 || ret2<0){
            ERR_LOG("set socket reuse address failed");
        }
    }
      //设置非阻塞模式
    void NonBlocking(){
        int flags = fcntl(_sockfd, F_GETFL, 0);
        if (flags == -1) {
            ERR_LOG("fcntl get flags failed");
            return;
        }
        if (fcntl(_sockfd, F_SETFL, flags | O_NONBLOCK) == -1) {
            ERR_LOG("fcntl set non-blocking failed");
        }
    }
};


class Poller;
class EventLoop;
//可以看出一个channel对象监管一个套接字
class Channel{
    private:
        int _fd;
        EventLoop* _loop;
        uint32_t _events;//要监控的事件
        uint32_t _revents;//实际发生的事件
        using EventCallback=std::function<void()>;
        EventCallback _read_cb;//读事件回调
        EventCallback _write_cb;//写事件回调
        EventCallback _error_cb;//错误事件回调
        EventCallback _close_cb;//链接断开事件回调
        EventCallback _event_cb;//任意事件回调
    public:
        ~Channel() {
            _read_cb = nullptr;
            _write_cb = nullptr;
            _error_cb = nullptr;
            _close_cb = nullptr;
            _event_cb = nullptr;
        }
        Channel(EventLoop*loop,int fd)
        :_loop(loop),_fd(fd),_events(0),_revents(0),
        _read_cb(NULL),_write_cb(NULL),_error_cb(NULL),_close_cb(NULL),_event_cb(NULL){}
        void SetRevents(uint32_t revents){_revents=revents;}
        uint32_t Events(){return _events;}
        int Fd(){return _fd;}
        void SetReadCallback(const EventCallback&cb){_read_cb=cb;}
        void SetWriteCallback(const EventCallback&cb){_write_cb=cb;}
        void SetErrorCallback(const EventCallback&cb){_error_cb=cb;}
        void SetCloseCallback(const EventCallback&cb){_close_cb=cb;}
        void SetEventCallback(const EventCallback&cb){_event_cb=cb;}
        //是否监控了可读
        bool ReadAble(){return _events&EPOLLIN;}
        //是否监控了可写
        bool WriteAble(){return _events&EPOLLOUT;}
        //启动读事件监控
        void EnableRead(){_events|=EPOLLIN;Update();}
        //启动写事件监控
        void EnableWrite(){_events|=EPOLLOUT;Update();}
        //关闭读事件监控
        void DisableRead(){_events&=~EPOLLIN;Update();}
        //关闭写事件监控
        void DisableWrite(){_events&=~EPOLLOUT;Update();}
        //关闭所有事件监控
        void DiableAll(){_events=0;Update();}
        //移除监控
        void Remove();
        void Update();
        //处理事件,一旦链接触发事件，就调用这个函数
        void HandleEvent(){
            if((_revents&EPOLLIN)||(_revents&EPOLLRDHUP)||(_revents&EPOLLPRI))
            {
                if(_read_cb)_read_cb();
            }
            //有可能释放链接的操作一次只处理一个
            if(_revents&EPOLLOUT)
            {
                if(_write_cb)_write_cb();
                //  刷新活跃度要在写操作完成之后，在写操作前刷新，如果写操作耗时很长，那么写完后直接就会被判断链接超时
                //    之后刷新，即是时间很长也可以立刻重新刷新活跃度              
            }
            else if(_revents&EPOLLERR)
            {
                if(_error_cb)_error_cb();//活跃度刷新在关闭链接之前，因为关闭连接后刷新活跃度没有意义
            }
            else if(_revents&EPOLLHUP)
            {
                if(_close_cb)_close_cb();
            }
            if(_event_cb)_event_cb();
        }

};



#define MAX_EPOLLEVENTS 1024//默认最大事件数
class Poller{
    private:
        int _epfd;
        struct epoll_event _evs[MAX_EPOLLEVENTS];
        std::unordered_map<int,Channel*> _channels;
        //根据op选项进行添加、修改或删除
        void Update(Channel*channel,int op){
            struct epoll_event ev;
            ev.data.fd=channel->Fd();
            ev.events=channel->Events();
            int ret=epoll_ctl(_epfd,op,channel->Fd(),&ev);
            if(ret<0){
                ERR_LOG("epoll_ctl failed");
                exit(1);
            }
        }
        //判断channel是否存放在poller中
        bool HasChannel(Channel*channel){
            return _channels.find(channel->Fd())!=_channels.end();
        }
    public:
        Poller(){
            _epfd=epoll_create(1);
            if(_epfd<0){
                ERR_LOG("epoll_create failed");
                exit(1);
            }
        }
        //添加或修改监控事件
        void UpdateChannel(Channel*channel){
            if(!HasChannel(channel))
            {
            _channels[channel->Fd()]=channel;
            Update(channel,EPOLL_CTL_ADD);
            }
            else{
                Update(channel,EPOLL_CTL_MOD);}
        }
        //移除监控事件
        void RemoveChannel(Channel*channel){
            if(HasChannel(channel)){
                Update(channel,EPOLL_CTL_DEL);
                _channels.erase(channel->Fd());
            }
        }
        //开始监控返回活跃链接
        void Poll(std::vector<Channel*>*active){
            //阻塞监控
            int ret=epoll_wait(_epfd,_evs,MAX_EPOLLEVENTS,-1);
            if(ret<0){
                if(errno!=EINTR)return;//被信号打断
                ERR_LOG("epoll_wait failed:%s",strerror(errno));
                exit(1);
            }
            //处理活跃事件
            for(int i=0;i<ret;i++){
            int fd=_evs[i].data.fd;
            auto it=_channels.find(fd);
            assert(it!=_channels.end());
            it->second->SetRevents(_evs[i].events);
            active->push_back(it->second);
            }
            return;
        }
};


using TaskFunc=std::function<void()>;//任务函数类型
using ReleaseFunc=std::function<void()>;
class TimerTask//定时器任务类
{
private:
    bool _is_deleted=false;//是否已经被删除
    uint64_t _id;//标识任务，方便查找
    uint32_t _timeout;//超时时间，每隔多长时间算一次超时
    TaskFunc _task_cb;//定时器对象要执行的定时任务
    ReleaseFunc _release;//用于删除TimerWheel保存的weakptr定时器对象
public:
    TimerTask(uint64_t id,uint32_t delay,const TaskFunc& task_cb)
    :_id(id),_timeout(delay),_task_cb(task_cb),_release(nullptr),_is_deleted(false){}
    //在析构时执行定时任务
    ~TimerTask(){
      if(!_is_deleted)
        _task_cb();
      _release();
    }
    //取消定时器任务
    void Cancel(){_is_deleted=true;}
    uint32_t DelayTime(){return _timeout;}//获取定时器的延迟时间
    void SetRelease(const ReleaseFunc& release){_release=release;}//设置释放函数
};

class EventLoop;

class TimerWheel{//时间轮类
  private:
    int _tick;//当前时间轮走到哪里了--秒针
    int _capacity;//时间轮的容量--最大延迟时间
    using WeakTask=std::weak_ptr<TimerTask>;//弱智能指针，指向定时器任务
    using PtrTask=std::shared_ptr<TimerTask>;//智能指针，指向定时器任务
    std::vector<std::vector<PtrTask>> _wheel;
    std::unordered_map<uint64_t,WeakTask> _timers;
    void RemoveTimer(uint64_t id){
        auto it=_timers.find(id);
        if(it!=_timers.end()){
          _timers.erase(it);
      }
    }
    int timefd;
    EventLoop*_loop;
    std::unique_ptr<Channel> _timer_channel;
        //添加定时任务
    void TimerAddInLoop(uint64_t id,uint32_t delay,const TaskFunc& task_cb){
            // auto task=std::make_shared<TimerTask>(id,delay,task_cb);
            // task->SetRelease(std::bind(&TimerWheel::RemoveTimer,this,id));
            // _timers[id]=WeakTask(task);
            // _wheel[(delay+_tick)%_capacity].push_back(task);
        }
        //刷新定时任务
        void TimerRefreshInLoop(uint64_t id){
        //     auto it=_timers.find(id);
        //     if(it!=_timers.end()){
        //       PtrTask pt=it->second.lock();
        //       uint32_t delay=pt->DelayTime();
        //       _wheel[(delay+_tick)%_capacity].push_back(pt);
        //   }
        }
         //取消定时器任务
    void TimerCancelInLoop(uint64_t id){
        auto it=_timers.find(id);
        if(it!=_timers.end()){
          PtrTask pt=it->second.lock();
          if(pt) {
            printf("finished cancel sucesses\n");
            pt->Cancel();
          }
        }
    }
  public:
        TimerWheel(EventLoop*loop):_capacity(1024),_tick(0),_wheel(_capacity),_loop(loop),timefd(CreatTimerFd())
        {
        _timer_channel=std::unique_ptr<Channel>(new Channel(loop,timefd));
        _timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime,this));
        _timer_channel->EnableRead();
        }
        uint64_t ReadTimerFd(){
        uint64_t value;
        int ret=read(timefd,&value,sizeof(value));
        if(ret<0){
            ERR_LOG("read timerfd failed");
            exit(1);
        }
        return value;
    }
    void OnTime(){
       int t= ReadTimerFd();
       for(int i=0;i<t;i++)
        RunTimerTask();
    }
    static int CreatTimerFd(){
        //创建timefd
        int ret=timerfd_create(CLOCK_MONOTONIC,0);
        if(ret<0){
            ERR_LOG("create timerfd failed");
            exit(1);
        }
        //设置timefd参数
        struct itimerspec its;
        its.it_value.tv_sec=1;//第一次超时时间为1s
        its.it_value.tv_nsec=0;
        its.it_interval.tv_sec=1;//第一次超时之后，每隔1s超时一次
        its.it_interval.tv_nsec=0;
        int ret2=timerfd_settime(ret,0,&its,NULL);
        if(ret2<0){
            ERR_LOG("set timerfd failed");
            exit(1);
        }
        return ret;
    }
    //保证在同一个线程完成，因为_times成员是公共资源，要保证线程安全
    void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& task_cb);
    //保证在同一个线程完成，因为_times成员是公共资源，要保证线程安全
    void TimerRefresh(uint64_t id);
    void TimerCancel(uint64_t id);
    //访问了公共资源，存在线程安全，不可以被外界使用者被调用，只能固定在eventloop线程执行，比如被eventloop对象调用
    bool HasTimer(uint64_t id){
        return _timers.find(id)!=_timers.end();
    }
    //这个函数应该每秒执行一次，
    void RunTimerTask(){
            _tick=(_tick+1)%_capacity;
        //clear会清理数组中的shared_ptr资源
        _wheel[_tick].clear();
    }
};



class EventLoop{
    private:
         int _event_fd;//通过它唤醒IO事件监控可能导致的阻塞
        std::thread::id _thread_id;//当前线程id
       
        std::unique_ptr<Channel> _event_channel;//_event_fd对应的Channel对象
        Poller _poller;
        using Functor=std::function<void()>;
        std::vector<Functor> _tasks;//任务池
        std::mutex _mutex;//任务池锁
        TimerWheel _timer_wheel;//定时器模块
    public:
        //执行所有任务池的任务
        void RunAllTask(){
            std::vector<Functor> functor;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _tasks.swap(functor);
            }
            for(auto &f:functor)
            {
                // 执行f三种回调
                // 1. Connection模块的ReleaseInLoop函数
                // 2. Connection模块的非活跃事件销毁
                // 3. Connection模块的启动事件监控
                // 崩溃在f执行上，一般来说就是Coonection实例对象被释放了，然后还有行为在执行上述回调，进而崩溃
                // printf("f: %p", f);
                    try {
                if (f) {  // 检查回调是否有效
                    f();
                }
                 } catch (const std::exception& e) {
                   std::cout << ("Callback error: %s", e.what()) << std::endl;;
                }
            }
           return;
        }

        //读事件回调
        void ReadEventFd(){
            uint64_t value;
            int ret=read(_event_fd,&value,sizeof(value));
            if(ret<0){
                //被信号打断              无数据可读
                if(errno==EINTR||errno==EAGAIN||errno==EWOULDBLOCK)return;
                ERR_LOG("read eventfd failed");
                exit(1);
            }
            return;
        }
        static int CreatEventFd(){
            int ret=eventfd(0,EFD_NONBLOCK|EFD_CLOEXEC);
            if(ret<0){
                ERR_LOG("create eventfd failed");
                exit(1);
            }
            return ret;
        }
        //唤醒EventLoop
        //就是给eventfd写一个数据，让它触发可读事件
        void WakeUpEventFd(){
            uint64_t value=1;
            int ret=write(_event_fd,&value,sizeof(value));
            if(ret<0){
                //被信号打断
                if(errno==EINTR)return;
                ERR_LOG("write eventfd failed");
                exit(1);
            }
            return;
        }
        EventLoop():_thread_id(std::this_thread::get_id())
                    ,_event_fd(CreatEventFd()),_event_channel(new Channel(this,_event_fd)),_timer_wheel(this){
                //设置_event_fd的读事件回调
                _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventFd,this));
                _event_channel->EnableRead();
        }
         //启动EventLoop
        //1----进行事件监控，2----处理事件，3----处理任务池中的任务
        void Start(){
            while(true){
            std::vector<Channel*> actives;
            _poller.Poll(&actives);//拿到有事件的channel
            //2----处理事件
            for(auto channel:actives)
            {
                channel->HandleEvent();
            }
            //3----处理任务池中的任务
            RunAllTask();
        }
        }
        //判断要执行的任务是否处于当前线程，如果是就执行否则压入任务池
        void RunInLoop(const Functor&cb){
            if(IsInLoop())
                return cb();
            else
                return QueueInLoop(cb);
        }
        //将任务压入任务池
        void QueueInLoop(const Functor&cb){
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _tasks.push_back(cb);
                //通知EventLoop有任务需要处理
            }
            WakeUpEventFd();
        }
        void AssertInloop(){
            assert(IsInLoop());
        }
        //判断当前线程是否是EventLoop所在线程
        bool IsInLoop(){return std::this_thread::get_id()==_thread_id;}
        //修改或添加描述符的事件监控
        void UpdateEvent(Channel*channel){ _poller.UpdateChannel(channel);}
        //移除描述符的事件监控
        void RemoveEvent(Channel*channel){ _poller.RemoveChannel(channel);}  
        //添加定时器任务
        void AddTimer(uint64_t id,uint32_t delay,const TaskFunc& task_cb){
            _timer_wheel.TimerAdd(id,delay,task_cb);
        }
        //刷新定时器任务
        void RefreshTimer(uint64_t id){
            _timer_wheel.TimerRefresh(id);
        }
        //取消定时器任务
        void CancelTimer(uint64_t id){
            _timer_wheel.TimerCancel(id);
        }
        //是否有某个定时器任务
        bool HasTimer(uint64_t id){return _timer_wheel.HasTimer(id);}
};



class LoopThread{
    private:
        //避免线程已创建、loop还未实例化时外界获取loop
        std::mutex _mutex;
        std::condition_variable _cond;
        // std::thread _thread; 
        // EventLoop* _loop;    //需要在线程内实例化


        EventLoop* _loop;    //需要在线程内实例化
         std::thread _thread; 
    
    private:
        //实例化EventLoop,唤醒cond上的线程，并开始运行EventLoop
        void ThreadEntry(){
            EventLoop loop;
            {
                std::unique_lock<std::mutex> lock(_mutex);//加锁
                _loop=&loop;
                _cond.notify_all();
            }
            loop.Start();
        }
    public:
        LoopThread():_loop(nullptr),_thread(std::thread(&LoopThread::ThreadEntry,this)){}
        EventLoop*GetLoop(){
            EventLoop* loop=nullptr;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _cond.wait(lock,[&]{return _loop!=nullptr;});
                loop=_loop;
            }
            return loop;
        } 
};


class LoopThreadPool{
    private:
        int _thread_num;
        int _next_idx;
        EventLoop* _base_loop;
        std::vector<LoopThread*> _threads;
        std::vector<EventLoop*> _loops;
    public:
        LoopThreadPool(EventLoop*base_loop):_thread_num(0),_next_idx(0),_base_loop(base_loop){}
        void SetThreadNum(int thread_num){_thread_num=thread_num;}
        //创建从属LoopThread
        void Creat(){
            if(_thread_num<=0) return;
            _threads.resize(_thread_num);
            _loops.resize(_thread_num);
            for(int i=0;i<_thread_num;i++){
                _threads[i]=new LoopThread();
                _loops[i]=_threads[i]->GetLoop();
            }
            return;           
        }
        EventLoop* NextLoop(){
            if(_thread_num==0)
            return _base_loop;
                _next_idx=(1+_next_idx)%_thread_num;
            return _loops[_next_idx];
        //return _loops[0];
        }
};
//           连接已关闭   正在建立连接  连接成功  正在断开连接
typedef enum{DISCONECTED,CONNEDTING,CONNECTED,DISCONNECTING} ConnState;
//所有对连接的操作都是基于Connect模块的，如果进行某操作时连接已经不存在就会出现内存访问错误
//因为我们使用智能指针管理Connect的生命周期，以规避上述风险
class Connection;
using PtrConnection=std::shared_ptr<Connection>; 
class Connection :public std::enable_shared_from_this<Connection>{
    private:
        //我们这里把每个连接对应的定时器任务id设置为_conn_id
        uint64_t _conn_id;//连接的唯一标识
        int _sockfd;//套接字描述符
        ConnState _state;//连接状态
        EventLoop* _loop;//连接所在的EventLoop
        Socket _socket;//封装的socket对象
        bool _enable_inactive_release;//是否允许释放不活跃的连接,true表示允许释放
        Channel _channel;//对链接的事件的管理
        Buffer _input_buffer;//输入缓冲区 --存放从socket中read的数据
        Buffer _output_buffer;//输出缓冲区 --存放需要write给socket的数据
        std::any _context;//请求上下文
        using ConnectedCallback=std::function<void(const PtrConnection&)>;
        ConnectedCallback _conn_cb;//连接成功后的回调函数
        using MessageCallback=std::function<void(const PtrConnection&,Buffer*)>;
        MessageCallback _msg_cb;//收到消息后的回调函数
        using ClosedCallback=std::function<void(const PtrConnection&)>;
        ClosedCallback _close_cb;//用户定义的连接断开后的回调函数 
        using AnyEventCallback=std::function<void(const PtrConnection&)>;
        AnyEventCallback _any_cb;//任意事件回调函数
        ClosedCallback _server_close_cb; //移除服务器内对该连接的管理信息
    private:
        //可读事件触发后调用的函数，将socket内的数据发送到接受缓冲区
        void HandleRead(){
            //1----从socket中读取数据放到缓冲区
            char buf[65536];//64kb
            //非阻塞读取
            int n=_socket.NonBlockingRecv(buf,sizeof(buf));
            //std::cout<<"recv data size:"<<n<<std::endl;
            if(n<0){
               return ShutDownInLoop();
            }
             _input_buffer.WriteAndPush(buf,n);
             //std::cout<<"recv data:"<<_input_buffer.ReadPosition()<<std::endl;
           // 2--调用messagecallback
            if(_input_buffer.ReadAbleSize()>0){
                _msg_cb(shared_from_this(),&_input_buffer);
            }

        }
        //可写事件触发后调用的函数，将发送缓冲区的数据发送到socket
        void HandleWrite(){
           
            //1----从发送缓冲区读取数据
            ssize_t ret=_socket.NonBlockingSend(_output_buffer.ReadPosition(),_output_buffer.ReadAbleSize());
            if(ret<0){
                if(_input_buffer.ReadAbleSize()>0)
                    _msg_cb(shared_from_this(),&_input_buffer);
                 return Release();
            } 
            _output_buffer.MoveReadOffset(ret);
            if(_output_buffer.ReadAbleSize()==0){
                _channel.DisableWrite();
                if(_state==DISCONNECTING)
                    return Release();
            }
            return;
        }
            
       
        //错误事件触发后调用的函数
        void HandleError(){
            return HandleClose();
        }
        //任意事件触发后调用的函数
        void HandleAny(){
            //刷新活跃度
            if(_enable_inactive_release)_loop->RefreshTimer(_conn_id);
            // 用户可能自定义添加了别的
            if(_any_cb)_any_cb(shared_from_this());
        }
        //断开连接事件触发后调用的函数
        void HandleClose(){
            if(_input_buffer.ReadAbleSize()>0)
            _msg_cb(shared_from_this(),&_input_buffer);
            return Release();
        }
        //连接建立好后，设置状态，启动读监控,调用ConnectedCallback
        void EstablishedInLoop(){
            assert(_state==CONNEDTING);
            _state=CONNECTED;
            //读事件监控
            _channel.EnableRead();
            //调用连接成功回调
            if(_conn_cb)_conn_cb(shared_from_this());
        }
        //不是真的发送了数据，只是把要发送的数据放到发送缓冲区，启动可写事件监控
        void SendInLoop(Buffer& buf){
            if(_state==DISCONECTED)return;
            _output_buffer.WriteBuffAndPush(buf);
            if(!_channel.WriteAble())_channel.EnableWrite();
        }
        //实际直接关闭链接的函数
        void ReleaseInLoop(){
            // printf("ReleaseInLoop: %p\n",&Connection::ReleaseInLoop);
            //1--修改状态
            _state=DISCONECTED;
            //2--移除监控
            _channel.Remove();
            //3--关闭socket
            _socket.Close();
            //4--取消定时器任务
            if(_loop->HasTimer(_conn_id))
                DisableInactiveReleaseInLoop();
            //5--调用关闭回调函数，先调用用户的回调函数在调用移除服务器管理信息的回调函数
            if(_close_cb)_close_cb(shared_from_this());
            if(_server_close_cb)_server_close_cb(shared_from_this());
        }

        //不是真的关闭而是判断是否有数据待处理或待发送
        void ShutDownInLoop(){
            if(_state==DISCONECTED)return;
            _state=DISCONNECTING;
            if(_input_buffer.ReadAbleSize()>0)
                _msg_cb(shared_from_this(),&_input_buffer);
            if(_output_buffer.ReadAbleSize()>0)
            {
                if(!_channel.WriteAble())_channel.EnableWrite();
            }
            if(_output_buffer.ReadAbleSize()==0)Release();

        }
        void EnableInactiveReleaseInLoop(int sec){
            _enable_inactive_release=true;
            //如果该定时任务已存在就刷新否则添加
            if(_loop->HasTimer(_conn_id))
                _loop->RefreshTimer(_conn_id);
            else
            _loop->AddTimer(_conn_id,sec,std::bind(&Connection::Release,this));
        }
        void DisableInactiveReleaseInLoop(){
            _enable_inactive_release=false;
            if(_loop->HasTimer(_conn_id))
            _loop->CancelTimer(_conn_id);
        }
        void UpgradeInLoop(const std::any& context,const ConnectedCallback& conn_cb,const MessageCallback& msg_cb,
            const ClosedCallback& close_cb,const AnyEventCallback& any_cb){
                _context=context;
                _conn_cb=conn_cb;
                _msg_cb=msg_cb;
                _close_cb=close_cb;
                _any_cb=any_cb;
            }
    public:
            //我们这里把每个连接对应的定时器任务id设置为_conn_id
        Connection(EventLoop*loop,uint64_t _conn_id,int sockfd):
            _conn_id(_conn_id),_sockfd(sockfd),_state(CONNEDTING),_loop(loop),_socket(sockfd),_enable_inactive_release(false),
            _channel(loop,sockfd),_conn_cb(nullptr),_msg_cb(nullptr),_close_cb(nullptr),_any_cb(nullptr),_server_close_cb(nullptr){
                _channel.SetCloseCallback(std::bind(&Connection::HandleClose,this));
                _channel.SetReadCallback(std::bind(&Connection::HandleRead,this));
                _channel.SetWriteCallback(std::bind(&Connection::HandleWrite,this));
                _channel.SetErrorCallback(std::bind(&Connection::HandleError,this));
                _channel.SetEventCallback(std::bind(&Connection::HandleAny,this));
                //不可以在这里开始读监控，因为定时任务还没有添加，添加定时任务这一步在EnableInactiveReleaseInLoop中实现
                //开启读监控在EstablishedInLoop中实现
            }
        ~Connection(){DBG_LOG("Connection %lu destructed %p", _conn_id, static_cast<void*>(this));}
        int Fd(){return _sockfd;}
        int Id(){return _conn_id;}
        Connection* Get(){return this;}
        //当前连接状态
        ConnState State(){return _state;}
        bool Connected(){return _state==CONNECTED;}
        //获取上下文,注意是指针
        std::any* Context(){return &_context;}
        //连接建立后，进行channel回调设置，启动读监控的
        void Established(){_loop->RunInLoop(std::bind(&Connection::EstablishedInLoop,this));}
        void SetConnectedCallBack(const ConnectedCallback& conn_cb){_conn_cb=conn_cb;}
        void SetMessageCallBack(const MessageCallback& msg_cb){_msg_cb=msg_cb;}
        void SetClosedCallBack(const ClosedCallback& close_cb){_close_cb=close_cb;}
        void SetAnyEventCallBack(const AnyEventCallback& any_cb){_any_cb=any_cb;}
        void SetServerCloseCallBack(const ClosedCallback& server_close_cb){_server_close_cb=server_close_cb;}
        void SetContext(const std::any& context){_context=context;}
        //发送数据到发送缓冲区，启动写事件监控
        void Send(const char*date,size_t len){
            //该函数不一定被立刻执行而是压入任务池，所以不要把date保存，防止之后他被释放
            Buffer buf;
            buf.WriteAndPush(date,len);
            _loop->RunInLoop(std::bind(&Connection::SendInLoop,this,std::move(buf)));
        }
        //关闭连接--不是直接关闭，而是判断有无数据待处理
        void ShutDown(){
            _loop->RunInLoop(std::bind(&Connection::ShutDownInLoop,this));
        }
        void Release(){
            _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop,this));
        }
        //开启非活跃销毁,并设置超过多长时间无通信就是超时
        void EnableInactiveRelease(int sec){
            // 就这一句，压入到队列中有问题
            _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop,this,sec));
        }
        //关闭非活跃销毁
        void DisableInactiveRelease(){
            _loop->RunInLoop(std::bind(&Connection::DisableInactiveReleaseInLoop,this));
        }
        //切换协议--重置上下文记忆阶段性处理函数,非线程安全
        //为了保证这个函数立刻被执行，他不能被压入任务池，而是直接执行，这就要求这个函数的执行本身就在Event_Loop线程中
        // 为了保证上述要求加入assert断言，如果不在该线程直接报错终止
        void Upgrade(const std::any &context,const ConnectedCallback& conn_cb,const MessageCallback& msg_cb,
            const ClosedCallback& close_cb,const AnyEventCallback& any_cb)
            {
                _loop->AssertInloop();
                _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop,this,context,conn_cb,msg_cb,close_cb,any_cb));
            }
};




class Acceptor{
    private:
        Socket _listen_socket;//监听套接字
        EventLoop* _loop;//监听所在的EventLoop
        Channel _channel;//监听套接字的事件管理
        using AccpetCallBack=std::function<void(int)>;
        AccpetCallBack _accpet_cb;//监听套接字的回调函数
    private:
        int CreateServer(int port){
            bool ret=_listen_socket.CreateServer(port);
            assert(ret==true);
            return _listen_socket.Fd();
        }
        //获取新连接，调用accept_cb
        void HandleRead(){
            int newfd=_listen_socket.Accept();
            if(newfd<0) return;
            if(_accpet_cb)
            _accpet_cb(newfd);
        }
    public:
    //不可以把启动读事件监控设置在构造函数，如果启动后，设置回调acceptor之前出现连接来访问，该连接无法被转化为一个connection且不会被销毁
    //造成资源浪费
        Acceptor(EventLoop*loop,int port):_loop(loop),_listen_socket(CreateServer(port)),_channel(loop,_listen_socket.Fd())
        {
            _channel.SetReadCallback(std::bind(&Acceptor::HandleRead,this));
        }
        void Listen(){
            _channel.EnableRead();
        }
        void SetAcceptCallBack(const AccpetCallBack& accpet_cb)
        {_accpet_cb=accpet_cb;}
};



class TcpServer{
    private:
        uint64_t _next_id;//自动增长的连接ID
        EventLoop _base_loop;//主线程的EventLoop,负责监听
        Acceptor _accpetor;//监听套接字的管理对象
        LoopThreadPool _pool;//线程池管理对象
        int _timeout;//多长时间无通信即是非活跃连接
        bool _enable_inactive_release;//是否允许释放不活跃的连接,true表示允许释放
        std::unordered_map<uint64_t,PtrConnection> _conns;//存放所有连接的容器
        using ConnectedCallback=std::function<void(const PtrConnection&)>;
        ConnectedCallback _conn_cb;//连接成功后的回调函数
        using MessageCallback=std::function<void(const PtrConnection&,Buffer*)>;
        MessageCallback _msg_cb;//收到消息后的回调函数
        using ClosedCallback=std::function<void(const PtrConnection&)>;
        ClosedCallback _close_cb;//用户定义的连接断开后的回调函数 
        using AnyEventCallback=std::function<void(const PtrConnection&)>;
        AnyEventCallback _any_cb;//任意事件回调函数
    private:
        //创建connection对象，并设置channel参数
        void NewConnection(int sockfd){
            std::cout <<  "sockfd: " <<  sockfd << std::endl;
           // printf("NewConnection: %p\n", &TcpServer::NewConnection);
            _next_id++;
            PtrConnection connection=std::make_shared<Connection>(_pool.NextLoop(),_next_id,sockfd);
            connection->SetMessageCallBack(_msg_cb);
            connection->SetServerCloseCallBack(std::bind(&TcpServer::RemoveConnection,this,std::placeholders::_1));
            connection->SetConnectedCallBack(_conn_cb);
            connection->SetClosedCallBack(_close_cb);
            connection->SetAnyEventCallBack(_any_cb);
            if(_enable_inactive_release)
                connection->EnableInactiveRelease(_timeout);
            connection->Established();
            //_conns[_next_id]=connection;
            _conns.insert(std::make_pair(_next_id, connection));
        }
        void RemoveConnectionInLoop(const PtrConnection& conn){
            int id=conn->Id();
            auto it=_conns.find(id);
            if(it!=_conns.end()){
                _conns.erase(it);
            }
        }
        void RemoveConnection(const PtrConnection& conn){
            _base_loop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop,this,conn));
        }
        //添加定时任务，在指定时间后执行task_cb
        void RunAfterInLoop(const TaskFunc& task_cb,int delay){
            _next_id++;
            _base_loop.AddTimer(_next_id,delay,task_cb);
        }
    public:
        TcpServer(int port):_next_id(0),_enable_inactive_release(false),
            _timeout(0),_conn_cb(nullptr),_msg_cb(nullptr),_close_cb(nullptr),_any_cb(nullptr),
            _pool(&_base_loop),_accpetor(&_base_loop,port){
                //不要再构造函数建立线程池，此时线程池的线程数量为0
                _accpetor.SetAcceptCallBack(std::bind(&TcpServer::NewConnection,this,std::placeholders::_1));
                _accpetor.Listen();//虽然开始监听但是_base_loop还没有启动，所以不用担心事件由于回调未设置错误响应
            }
        void SetThreadCount(int thread_num){_pool.SetThreadNum(thread_num);}
        void SetConnectedCallBack(const ConnectedCallback& conn_cb){_conn_cb=conn_cb;}
        void SetMessageCallBack(const MessageCallback& msg_cb){_msg_cb=msg_cb;}
        void SetClosedCallBack(const ClosedCallback& close_cb){_close_cb=close_cb;}
        void SetAnyEventCallBack(const AnyEventCallback& any_cb){_any_cb=any_cb;}
        void EnableInactiveRelease(int timeout){_timeout=timeout;_enable_inactive_release=true;}
        void DisableInactiveRelease(){_enable_inactive_release=false;}
        //添加定时任务，在指定时间后执行task_cb
        void RunAfter(const TaskFunc& task_cb,int delay){
            _base_loop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,task_cb,delay));}
        void Start(){
            _pool.Creat();//创建子线程
            _base_loop.Start();
        }
};
void Channel::Remove(){_loop->RemoveEvent(this);}
void Channel::Update(){_loop->UpdateEvent(this);}
 //保证在同一个线程完成，因为_times成员是公共资源，要保证线程安全
 void TimerWheel::TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& task_cb){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop,this,id,delay,task_cb));
}
    //保证在同一个线程完成，因为_times成员是公共资源，要保证线程安全
void TimerWheel::TimerRefresh(uint64_t id){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop,this,id));
}
void TimerWheel::TimerCancel(uint64_t id){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop,this,id));
}



class NetWork{
    public:
    // SIGPIPE 信号：SIGPIPE 信号一般在向一个已经关闭的管道或者套接字写数据时产生。默认情况下，进程接收到 SIGPIPE 信号会终止。
    // SIG_IGN：SIG_IGN 是一个预定义的常量，用于表示忽略信号。所以 signal(SIGPIPE, SIG_IGN); 这行代码的作用是让程序忽略 SIGPIPE 信号，避免程序因为向已关闭的套接字写数据而终止。
    //我们无须担心连接关闭后继续send导致触发异常导致关闭程序    
    NetWork(){
            DBG_LOG("NetWork init success");
            signal(SIGPIPE,SIG_IGN);
        }
};
static NetWork nw;